package com.xiaochong.tir.common.util;

import com.alibaba.fastjson.JSON;
import com.xiaochong.tir.common.exception.fs.ReflectionException;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.util.Bytes;

import java.util.Map;

/**
 * hbase 工具类
 * Created by cx on 2018/4/24.
 */
public class HbaseUtils {

    /**
     * 插入数据
     * @param columnFamily 列族
     * @param obj 对象
     * @param put
     * @param attributes 不put的数据
     * @param falg 是否要反射父类的数据
     * @throws ReflectionException
     */
    public static void  addColumn(String columnFamily, Object obj, Put put, Map<String, String> attributes, boolean falg) throws ReflectionException {

        Map<String, Object> map;
        try{
            if(falg){
                map = ObjectDynamicCreatorUtil.getFieldVlaueAndSupper(obj);
            }else{
                map = ObjectDynamicCreatorUtil.getFieldVlaue(obj);
            }
        }catch(Exception e){
            throw new ReflectionException("反射出现异常", e);
        }


        for(Map.Entry<String, Object> entry: map.entrySet()){
            if(!attributes.containsKey(entry.getKey())){
                put.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(entry.getKey()), Bytes.toBytes(entry.getValue().toString()));
            }


        }

    }

    /**
     *   将LIst 转成json格式的数据插入hbase
     * @param columnFamily 列族
     * @param obj 对象
     * @param put
     * @param attributes 不put的数据
     * @throws ReflectionException
     */
    public static void  addArrayColumn(String columnFamily, Object obj,Put put,Map<String, String> attributes) throws ReflectionException{

        Map<String, Object> map;
        try{
            map = ObjectDynamicCreatorUtil.getFieldVlaue(obj);

        }catch(Exception e){
            throw new ReflectionException(e.getMessage(), e);
        }


        for(Map.Entry<String, Object> entry: map.entrySet()){
            if(!attributes.containsKey(entry.getKey())){
                put.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(entry.getKey()), JSON.toJSONBytes(entry.getValue()));
            }

        }

    }
}
